package netty.rpc.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import netty.rpc.common.exceptions.NettyRpcReadTimeOutException;
import netty.rpc.common.utils.DateUtil;
import netty.rpc.core.NettyRpcServiceChannelFutureHolder;
import netty.rpc.core.RpcDecoder;
import netty.rpc.core.RpcEncode;
import netty.rpc.server.RpcResponse;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * @author Jim
 * @version 1.0.0
 * @descripition:
 * @date 2022/5/14 11:28
 */
public class NettyRpcClient {

    private Logger logger = LogManager.getLogger(NettyRpcClient.class);

    private InterfaceConfig interfaceConfig;

    private Long timeOut;

    private NettyRpcClientHandler nettyRpcClientHandler;

    public NettyRpcClient(InterfaceConfig interfaceConfig) {
        this.interfaceConfig = interfaceConfig;
        this.timeOut = interfaceConfig.getTimeOut();
        this.nettyRpcClientHandler = new NettyRpcClientHandler(timeOut);
    }

    /**
     * 连接目标服务器
     */
    public ChannelFuture connect(InterfaceConfig interfaceConfig){
        logger.info("#NettyRpcClient# try to connect service:{}",interfaceConfig.getServiceName());
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast(new RpcEncode(RpcRequest.class))
                                .addLast(new RpcDecoder(RpcResponse.class))
                                .addLast(new NettyRpcTimeOutResponseHandler())
                                .addLast(nettyRpcClientHandler);
                    }
                });
        try {

            ChannelFuture channelFuture = bootstrap.connect(interfaceConfig.getHost(),interfaceConfig.getPort()).sync();
            NettyRpcServiceChannelFutureHolder.put(interfaceConfig.getServiceName(), channelFuture);
            logger.info("#NettyRpcClient# {} connected successfully",interfaceConfig.getServiceName());
            return channelFuture;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


    public RpcResponse remoteCall(RpcRequest rpcRequest) throws Throwable {
        logger.info("#NettyRpcClient# remote call service:{} time:{}",
                interfaceConfig.getServiceName(), DateUtil.getCurrentTime("yyyy-MM-dd HH:mm:sss"));
        ChannelFuture channelFuture = NettyRpcServiceChannelFutureHolder.get(interfaceConfig.getServiceName());
        if (channelFuture == null){
            channelFuture = connect(interfaceConfig);
            NettyRpcServiceChannelFutureHolder.put(interfaceConfig.getServiceName(), channelFuture);
        }
        NettyRpcRequestTimeHolder.markRequest(rpcRequest.getRequestId(), timeOut);
        channelFuture.channel().writeAndFlush(rpcRequest).sync();
        RpcResponse response = nettyRpcClientHandler.getResponse(rpcRequest.getRequestId());

        if (response.isSuccess()){
            return response;
        }

        throw new NettyRpcReadTimeOutException("remote call service: " + interfaceConfig.getServiceName() + " time out. request = " + rpcRequest);


    }
}
